package tech.yunjing.log.mq;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import tech.yunjing.log.dao.ApplicationLogDao;

import java.util.Optional;

/**
 * 日志数据消费者
 * @author miou
 *
 */
public class LogListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(LogListener.class);

    @Autowired
    private ApplicationLogDao applicationLogDao;

    /**
     * @param record 消息实体bean
     */
    @KafkaListener(topics = "log-service")
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            try {
                JSONObject jsonObject = JSON.parseObject(message.toString());
                applicationLogDao.save(jsonObject);
                System.out.println(message.toString());
            } catch (Exception e) {
                LOGGER.error("将接收到的消息保存到数据库时异常, 消息:{}, 异常:{}",message.toString(),e);
            }
        }
    }

}
